package com.guojizhuang.topn;

import com.guojizhuang.writable.MerchantInfoWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Comparator;
import java.util.TreeMap;

public class TopNReducer extends Reducer<LongWritable, Text,NullWritable, Text> {

    @Override
    protected void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        TreeMap<Long, String> treeMap= new TreeMap<>(new Comparator<Long>() {
            @Override
            public int compare(Long o1, Long o2) {
                return (int) (o1-o2);
            }
        });
        for (Text value1 : values)
        {
            String s = value1.toString();
            String[] value =s.split("\t");
            treeMap.put(((Long.parseLong(value[2]))+ Long.parseLong(value[3])),s);
            if(treeMap.size() >5){
                treeMap.remove(treeMap.firstKey());
            }
        }
        for(String value: treeMap.values()){
            context.write(NullWritable.get(),new Text(value));
        }
    }

}
